package com.csx.governance.traffic.mq.rocketmq.hook;

import cn.hutool.core.util.StrUtil;
import com.csx.governance.traffic.common.exception.TrafficGovernanceRuleConfigException;
import com.csx.governance.traffic.core.constants.TrafficAttributeConstants;
import com.csx.governance.traffic.core.metadata.ServiceMetadataLocator;
import com.csx.governance.traffic.route.model.swimlane.SwimLaneDefinition;
import com.csx.governance.traffic.route.swimlane.SwimLaneGovernanceRuleLocator;
import org.apache.rocketmq.client.hook.ConsumeMessageContext;
import org.apache.rocketmq.client.hook.ConsumeMessageHook;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.List;

/**
 * 消费消息的钩子
 * @author cuisongxu
 * @date 2023/2/28 16:24
 */
public class TrafficConsumeMessageHook implements ConsumeMessageHook {

    private final Logger logger = LoggerFactory.getLogger(TrafficConsumeMessageHook.class);

    private final SwimLaneGovernanceRuleLocator swimLaneGovernanceRuleLocator;

    private final String localApplicationName;

    private final ServiceMetadataLocator serviceMetadataLocator;

    public TrafficConsumeMessageHook(SwimLaneGovernanceRuleLocator swimLaneGovernanceRuleLocator,
                                     String localApplicationName,
                                     ServiceMetadataLocator serviceMetadataLocator) {
        this.swimLaneGovernanceRuleLocator = swimLaneGovernanceRuleLocator;
        this.localApplicationName = localApplicationName;
        this.serviceMetadataLocator = serviceMetadataLocator;
    }

    @Override
    public String hookName() {
        return "TrafficConsumeMessageHook";
    }

    @Override
    public void consumeMessageBefore(ConsumeMessageContext context) {
        List<MessageExt> msgList = context.getMsgList();
        Iterator<MessageExt> iterator = msgList.iterator();
        while(iterator.hasNext()) {
            MessageExt message = iterator.next();
            String swimLaneId = message.getUserProperty(TrafficAttributeConstants.SWIM_LANE);
            if (StrUtil.isNotEmpty(swimLaneId)) {
                SwimLaneDefinition.SwimLaneSubDefinition swimLaneSubDefinition = swimLaneGovernanceRuleLocator.getSwimLaneSubDefinition(swimLaneId, localApplicationName);

                // 如果泳道定义不为空,说明当前泳道存在这个消息的消费者，需要进一步判断当前实例是否在当前泳道中,如果当前实例是在当前泳道中则消费消息,否则不消费消息
                // 如果泳道定义为空, 说明当前泳道不存在这个消息的消费者, 需要找到这个泳道对应的基线泳道,
                //      情况一: 当前泳道存在对应的基线泳道,需要进一步判断基线泳道是否存在当前应用的定义
                //          ① 基线泳道中存在当前应用的定义, 如果是则进一步判断当前实例是否在基线泳道中,在则消费消息,否则不消费消息
                //          ② 基线泳道中不存在当前应用的定义,不消费该消息,并给出用户一个友好的提示，说明基线泳道不存在该应用,用户需要修改基线泳道定义
                //      情况二: 当前泳道不存在对应的基线泳道,不消费该消息
                if(swimLaneSubDefinition != null) {
                    String swimLaneKey = swimLaneSubDefinition.getSwimLaneKey();
                    String swimLaneValue = swimLaneSubDefinition.getSwimLaneValue();
                    // 当前实例是不在当前泳道中,丢弃消息
                    if(!swimLaneValue.equals(serviceMetadataLocator.get(swimLaneKey))) {
                        iterator.remove();
                    }
                }else {
                    String baseSwimLaneId = swimLaneGovernanceRuleLocator.getBaseSwimLaneIdBySwimLaneId(swimLaneId);

                    // 基线不存在,说明是因为当前泳道被删除了(因为非基线泳道必须存在对应的基线),此时默认 全局的基线泳道 作为其基线(注意: 正常情况下在灰度消息没有被消费完成之前,不应该删除其泳道)
                    if(StrUtil.isEmpty(baseSwimLaneId)) {
                        logger.error("swimLane [ " + swimLaneId + " ] don't have been deleted, please check governance config !");
                        baseSwimLaneId = swimLaneGovernanceRuleLocator.getGlobalBaseSwimLaneId();
                    }

                    if(baseSwimLaneId == null) {
                        throw new TrafficGovernanceRuleConfigException("global base swimLane not exist, please check governance config !");
                    }

                    // 基线存在但是基线的泳道定义不包含当前应用
                    SwimLaneDefinition.SwimLaneSubDefinition baseSwimLaneDefinition = swimLaneGovernanceRuleLocator.getSwimLaneSubDefinition(baseSwimLaneId, localApplicationName);
                    if(baseSwimLaneDefinition == null) {
                        logger.error("swimLane [ " + swimLaneId + " ] ' s baseSwimLane [" + baseSwimLaneId + "] 's swimLaneDefinitions don't have current application [ " + localApplicationName + " ], please check governance config !");
                        iterator.remove();
                        continue;
                    }
                    String swimLaneKey = baseSwimLaneDefinition.getSwimLaneKey();
                    String swimLaneValue = baseSwimLaneDefinition.getSwimLaneValue();

                    // 当前实例是不在基线泳道中,丢弃消息
                    if(!swimLaneValue.equals(serviceMetadataLocator.get(swimLaneKey))) {
                        iterator.remove();
                    }
                }
            }


        }
        System.out.println("TrafficConsumeMessageHook#consumeMessageBefore");
    }

    @Override
    public void consumeMessageAfter(ConsumeMessageContext context) {


        System.out.println("TrafficConsumeMessageHook#consumeMessageAfter");
    }
}
